package org.atlantis.evnlib.task;

import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import org.atlantis.evnlib.exception.EventException;
import org.atlantis.evnlib.execution.UnnamedThreadFactory;
import org.atlantis.evnlib.execution.rejecter.RejectedTaskHandler;

/**
 * 任务处理器。将每种任务逻辑划分，任务之间线程不会互相干扰，因此避免出现线程饿死。
 * 
 * @author <a href="mailto:seenoevil.cn@gmail.com">李曦</a><br/>
 *         Created at: 2011-6-30 下午09:19:35
 */
public final class TaskExecutor {

	private final RejectedTaskHandler rejectedTaskHandler;
	private final ExecutorService realExec;
	private final ConcurrentHashMap<String, TaskPool> taskPoolMap;
	private final AtomicBoolean closed = new AtomicBoolean();

	/**
	 * 创建任务处理器。任务处理器委托ThreadPoolExecutor来进行任务处理。
	 * 
	 * @param threadSize
	 */
	public TaskExecutor(RejectedTaskHandler rejectedTaskHandler) {
		this.realExec = Executors.newCachedThreadPool(UnnamedThreadFactory.create("TaskExecutor:Unnamed"));
		this.taskPoolMap = new ConcurrentHashMap<String, TaskPool>();
		this.rejectedTaskHandler = rejectedTaskHandler;
	}

	/**
	 * 关闭任务处理器。任务处理器关闭后，不能再接受新的任务。
	 */
	public Collection<Runnable> shutdownNow() {
		if (!closed.compareAndSet(false, true)) return null;
		List<Runnable> tasks = new LinkedList<Runnable>();
		for (TaskPool pool : taskPoolMap.values())
			tasks.addAll(pool.close());
		taskPoolMap.clear();
		realExec.shutdownNow();
		return tasks;
	}

	/**
	 * 创建任务池。
	 * 
	 * @param taskName
	 *            任务池名称
	 * @param fixedQuotas
	 *            固定线程数
	 * @param flexQuotas
	 *            弹性线程数
	 * @param taskQueueSize
	 *            任务队列的最大容量
	 * @param rejectedTaskHandler
	 *            当任务被拒绝时的处理器
	 */
	public void createTaskPool(String taskName, int fixedQuotas, int flexQuotas, int taskQueueSize) {
		checkShutdown();
		if (fixedQuotas <= 0 || flexQuotas < 0) throw new EventException("Quatos must > 0: %s.", taskName);
		if (taskQueueSize <= 0) throw new EventException("taskQueueSize must > 0: %s.", taskName);
		if (taskPoolMap.containsKey(taskName)) throw new EventException("TaskPool<%s> already exist.", taskName);
		TaskPool taskPool = new TaskPool(realExec, taskName, fixedQuotas, flexQuotas, rejectedTaskHandler, taskQueueSize);
		taskPoolMap.put(taskName, taskPool);
	}

	/**
	 * 销毁任务池。
	 * 
	 * @param taskName
	 *            任务池名称。如果该任务池不存在，则直接返回，不会给出提示信息。
	 */
	public Collection<Runnable> destoryTaskPool(String taskName) {
		TaskPool taskPool = taskPoolMap.remove(taskName);
		if (taskPool == null) return null;
		return taskPool.close();
	}

	private void checkShutdown() {
		if (closed.get()) throw new EventException("TaskExecutor is shutdown.");
	}

	/**
	 * 提交任务。提交任务到指定名称的任务池，如果该任务池不存在则抛出异常。
	 * 
	 * @param taskName
	 *            任务池名称
	 * @param task
	 *            任务
	 */
	public void submit(String taskName, Runnable task) {
		checkShutdown();
		TaskPool taskPool = taskPoolMap.get(taskName);
		if (taskPool == null) new EventException("No TaskPool<%s> found.", taskName);
		taskPool.submit(task);
	}

	/**
	 * 检查任务处理器是否关闭。
	 * 
	 * @return
	 */
	public boolean isShutdown() {
		return closed.get();
	}
}
